#ifndef MTL_EXECUTOR_HPP
#define MTL_EXECUTOR_HPP

#include <map>
#include "thread.hpp"
#include "future.hpp"
#include "mutex.hpp"
#include "queue.hpp"
#include "memory.hpp"

namespace mtl {

  class fixed_thread_pool {
    typedef std::map<thread::id, shared_ptr<thread> > thread_map;
    typedef blocking_queue<shared_ptr<runnable> >     task_queue;

    enum state {
      RUNNING  = 1,
      SHUTDOWN = 2,
    };

  public:
    template<class _Bind>
      shared_ptr<future<typename _Bind::result_type> > submit(_Bind b) {
        if (is_shutdown()) {
          throw std::runtime_error("pool is shutdowned.");
        }
        typedef typename _Bind::result_type result_type;
        typedef shared_ptr<runnable_future<result_type> > shared_runnable_future;
        shared_runnable_future ftr(new future_task<_Bind, result_type>(b));
        put_task(ftr);
        return ftr;
      }

  protected:
    void put_task(shared_ptr<runnable> r) {
      m_q.put(r);
      if (m_current_pool_size < m_core_pool_size) {
        add_thread();
      }
    }

    void add_thread() {
      lock_guard<mutex> lock(m_main_mtx);
      shared_ptr<thread> th(new thread(
                                  bind(&fixed_thread_pool::run, this)));
      m_ths.insert(thread_map::value_type(th->get_id(), th));
      ++m_current_pool_size;
    }

    shared_ptr<thread> get_thread(const thread::id& id) {
      lock_guard<mutex> lock(m_main_mtx);
      thread_map::iterator it = m_ths.find(id);
      if (it == m_ths.end()) {
        return shared_ptr<thread>();
      }
      return it->second;
    }

  protected:
    void run() {
      pthread_cleanup_push(delete_current_thread, (void*)this);
        while (true) {
          this_thread::check_cancel();
          shared_ptr<runnable> r = m_q.take();
          this_thread::check_cancel();
          if (r.get() != NULL) {
            r->run();
          }
        }
      pthread_cleanup_pop(1);
    }

    static void delete_current_thread(void* p) {
      fixed_thread_pool* owner = reinterpret_cast<fixed_thread_pool*>(p);
      if (owner->is_shutdown()) {
        return;
      }
      lock_guard<mutex> lock(owner->m_main_mtx);
      if (owner->is_shutdown()) {
        return;
      }
      shared_ptr<thread> th = owner->get_thread(this_thread::get_id());
      if (th) {
        owner->m_ths.erase(th->get_id());
        th.reset();
        --(owner->m_current_pool_size);
        if (!owner->is_shutdown() && owner->m_current_pool_size < owner->m_core_pool_size) {
          owner->add_thread();
        }
      }
    }

  public:
    explicit fixed_thread_pool(const int num)
      : m_state(RUNNING), m_current_pool_size(0), m_core_pool_size(num)
    {}

    ~fixed_thread_pool() {
      try {
        shutdown();
      } catch (...) {
        // ignore
      }
    }

    void shutdown() {
      if (is_shutdown()) {
        return;
      }
      lock_guard<mutex> lock(m_main_mtx);
      if (is_shutdown()) {
        return;
      }
      m_state = SHUTDOWN;
      if (!m_q.empty()) {
        m_q.clear();
      }
      if (!m_ths.empty()) {
        for (thread_map::iterator it=m_ths.begin(); it!=m_ths.end(); ++it) {
          shared_ptr<thread> th = it->second;
          th->interrupt();
          th->join();
          th.reset();
        }
        m_ths.clear();
      }
      m_current_pool_size = 0;
    }

    bool is_shutdown() const {
      return (m_state & SHUTDOWN) != 0;
    }

  private:
    recursive_mutex m_main_mtx;

    thread_map  m_ths;
    task_queue  m_q;

    volatile state  m_state;

    volatile int  m_current_pool_size;
    volatile int  m_core_pool_size;
  };

}
#endif

